package com.chukun.gateway.registry.nacos;

import com.alibaba.fastjson2.JSON;
import com.alibaba.nacos.api.PropertyKeyConst;
import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.api.naming.NamingFactory;
import com.alibaba.nacos.api.naming.NamingMaintainFactory;
import com.alibaba.nacos.api.naming.NamingMaintainService;
import com.alibaba.nacos.api.naming.NamingService;
import com.alibaba.nacos.api.naming.listener.Event;
import com.alibaba.nacos.api.naming.listener.EventListener;
import com.alibaba.nacos.api.naming.listener.NamingEvent;
import com.alibaba.nacos.api.naming.pojo.Instance;
import com.alibaba.nacos.api.naming.pojo.ListView;
import com.alibaba.nacos.api.naming.pojo.Service;
import com.alibaba.nacos.api.naming.pojo.ServiceInfo;
import com.chukun.gateway.common.constants.GatewayConst;
import com.chukun.gateway.common.definition.ServiceDefinition;
import com.chukun.gateway.common.instance.ServiceInstance;
import com.chukun.gateway.registry.api.RegisterCenter;
import com.chukun.gateway.registry.api.RegisterCenterListener;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;

import java.util.*;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @author chukun
 * @version 1.0.0
 * @description 基于nacos实现的注册中心
 * @createTime 2023/11/26 20:51
 */
@Slf4j
public class NacosRegisterCenter implements RegisterCenter {

    private static final String NAME_SPACE="gateway";

    private String registryAddress;

    private String env;

    // 主要用于维护服务实例信息
    private NamingService namingService;
    // 主要用于维护服务定义信息
    private NamingMaintainService namingMaintainService;
    // 监听器的列表
    private List<RegisterCenterListener> registerCenterListeners = new ArrayList<>();

    private static final ScheduledThreadPoolExecutor NACOS_SCHEDULED_POOL = new ScheduledThreadPoolExecutor(1, r -> {
        Thread t = new Thread(r);
        t.setDaemon(true);
        t.setName("nacos-register-service");
        return t;
    });

    @Override
    public void init(String registerAddress, String env) {

        this.registryAddress = registerAddress;
        this.env = env;
        try {
            Properties properties = new Properties();
            properties.put(PropertyKeyConst.SERVER_ADDR, registerAddress);
            properties.put(PropertyKeyConst.NAMESPACE, NAME_SPACE);
            this.namingService = NamingFactory.createNamingService(properties);
            this.namingMaintainService = NamingMaintainFactory.createMaintainService(properties);
        } catch (NacosException e) {
            log.error("NacosRegisterCenter.init.error ", e);
            throw new RuntimeException(e);
        }

    }

    @Override
    public void register(ServiceDefinition serviceDefinition, ServiceInstance serviceInstance) {

        try {
            // 构造nacos的实例
            Instance nacosInstance = new Instance();
            nacosInstance.setInstanceId(serviceInstance.getServiceInstanceId());
            nacosInstance.setPort(serviceInstance.getPort());
            nacosInstance.setIp(serviceInstance.getIp());
            nacosInstance.setMetadata(Map.of(GatewayConst.META_DATA_KEY, JSON.toJSONString(serviceInstance)));

            // 注册实例
            namingService.registerInstance(serviceDefinition.getServiceId(), env, nacosInstance);

            //更新服务定义
            namingMaintainService.updateService(serviceDefinition.getServiceId(), env, 0, Map.of(GatewayConst.META_DATA_KEY, JSON.toJSONString(serviceDefinition)));

            log.info("NacosRegisterCenter.register {} {}", serviceDefinition, serviceInstance);
        } catch (NacosException e) {
            log.error("NacosRegisterCenter.register.error ", e);
            throw new RuntimeException(e);
        }

    }

    @Override
    public void deregister(ServiceDefinition serviceDefinition, ServiceInstance serviceInstance) {
        try {
            // 下线服务
            namingService.deregisterInstance(serviceDefinition.getServiceId(),
                    env, serviceInstance.getIp(), serviceInstance.getPort());

            log.info("NacosRegisterCenter.deregister {} {}", serviceDefinition, serviceInstance);
        }catch (NacosException e) {
            log.info("NacosRegisterCenter.deregister.error ", e);
            throw new RuntimeException(e);
        }
    }

    @Override
    public void subscribeAllServices(RegisterCenterListener registerCenterListener) {
        // 添加到监听器列表
        registerCenterListeners.add(registerCenterListener);
        // 执行订阅所有服务
        this.doSubscribeAllServices();

        // 定时扫描
        NACOS_SCHEDULED_POOL.scheduleWithFixedDelay(this::doSubscribeAllServices, 1, 5, TimeUnit.SECONDS);
    }

    private void doSubscribeAllServices() {
        try {
            Set<String> subscribeServices = namingService.getSubscribeServices().stream().map(ServiceInfo::getName).collect(Collectors.toSet());
            if (CollectionUtils.isEmpty(subscribeServices)) {
                return;
            }
            int pageNo = 1;
            int pageSize = 100;
            ListView<String> servicesOfServer = namingService.getServicesOfServer(pageNo, pageSize, env);

            int totalCount = servicesOfServer.getCount();
            List<String> serviseList = servicesOfServer.getData();

            if (totalCount == 0) {
                return;
            }
            EventListener eventListener = new NacosRegisterListener();
            do {
                log.info("NacosRegisterCenter.doSubscribeAllServices.service list size {}", serviseList.size());

                for (String service : serviseList) {
                    if (subscribeServices.contains(service)) {
                        continue;
                    }
                    namingService.subscribe(service, eventListener);
                    log.info("NacosRegisterCenter.doSubscribeAllServices.subscribe {} {}", service, env);
                }
                serviseList = namingService.getServicesOfServer(++pageNo, pageSize, env).getData();
            } while (!CollectionUtils.isEmpty(serviseList) && serviseList.size() >= pageSize);
        } catch (NacosException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * nacos注册的监听器
     */
    public class NacosRegisterListener implements EventListener {

        @Override
        public void onEvent(Event event) {
          if (event instanceof NamingEvent) {
              NamingEvent namingEvent = (NamingEvent) event;
              String serviceName = namingEvent.getServiceName();

              try {
                  //获取服务定义信息
                  Service service = namingMaintainService.queryService(serviceName, env);
                  ServiceDefinition serviceDefinition = JSON.parseObject(service.getMetadata().get(GatewayConst.META_DATA_KEY), ServiceDefinition.class);

                  //获取服务实例信息
                  List<Instance> allInstances = namingService.getAllInstances(serviceName, env);
                  Set<ServiceInstance> serviceInstances = new HashSet<>();

                  for (Instance instance : allInstances) {
                      ServiceInstance serviceInstance = JSON.parseObject(instance.getMetadata()
                              .get(GatewayConst.META_DATA_KEY), ServiceInstance.class);
                      serviceInstances.add(serviceInstance);
                  }

                  registerCenterListeners.forEach(listener -> listener.onChange(serviceDefinition, serviceInstances));
              } catch (NacosException e) {
                  throw new RuntimeException(e);
              }
          }
        }
    }
}
